/*
 * simplemq_redis.hpp
 *
 *  Created on: 2020年4月4日
 *      Author: guyadong
 */
#ifdef SIMPLEMQ_MQTT_SIMPLEMQ_MQTT_HPP_
#error simplemq_mqtt.hpp included,cant include simplemq_redis.hpp at the same time
#error (不允许同时 include simplemq_mqtt.hpp和simplemq_redis.hpp)
#endif 
#ifndef SIMPLEMQ_REDIS_SIMPLEMQ_REDIS_HPP_
#define SIMPLEMQ_REDIS_SIMPLEMQ_REDIS_HPP_
#include <map>
#include <mutex>
#include <sstream>
#include <limits>
#include <simplemq/simplemq.hpp>
#include <winsock_initializer.h>
#include <uri_utils.hpp>
#include <simplemq/simplemq_commons.hpp>
#include <resource_manager.hpp>
#include <cpp_redis/cpp_redis>
namespace simplemq {

	/**
	 * 根据指定的 host/port/password 创建 redis uri 实例
	 * @param host host to be connected to
	 * @param port port to be connected to
	 * @param password password for redis authorization
	 * @param timeout_ms
	 * @param max_reconnects
	 * @param reconnect_interval_ms
	 * @return uri instance
	 */
	inline uri make_uri(const std::string &host, size_t port,
		const std::string &password = "",
		std::uint32_t timeout_ms = std::numeric_limits<std::uint32_t>::max(),
		std::int32_t max_reconnects = std::numeric_limits<std::uint32_t>::max(),
		std::uint32_t reconnect_interval_ms = std::numeric_limits<std::uint32_t>::max()) {
		return make_uri(REDIS_SCHEMA, host, port, "", password, timeout_ms, max_reconnects, reconnect_interval_ms);
	}

	/**
	 * 对uri对象进行归一化，输出标准的 redis uri(redis://:password@host:port),不包含请求参数
	 * @param uri
	 * @return uri instance
	 */
	inline uri normalize(const uri &uri) {
		return make_uri(uri.get_host(), uri.get_port(), uri.get_password());
	}
	struct mq_client_params : public base_mq_client_params {
		mq_client_params(
			const std::string &host = DEFAULT_HOST,
			int port = DEFAULT_REDIS_PORT,
			const std::string &password = DEFAULT_PASSWORD,
			const std::string &username = DEFAULT_USERNAME,
			std::uint32_t timeout_ms = DEFAULT_TIMEOUT_MS,
			std::int32_t max_reconnects = DEFAULT_MAX_RECONNECTS,
			std::uint32_t reconnect_interval_ms = DEFAULT_RECONNECT_INTERVAL_MS)
			: base_mq_client_params(REDIS_SCHEMA,DEFAULT_REDIS_PORT,host, port, password, DEFAULT_USERNAME, timeout_ms, DEFAULT_MAX_RECONNECTS, reconnect_interval_ms) {}
		mq_client_params(const uri&uri) : base_mq_client_params(REDIS_SCHEMA, DEFAULT_REDIS_PORT, parse_components(uri)) {}
		_BASIC_JSON_TPL_DECLARATION_
		mq_client_params(const _BASIC_JSON_TPL_ &params) : base_mq_client_params(REDIS_SCHEMA, DEFAULT_REDIS_PORT, params) {}
		/** 输出标准的 redis uri(redis://:password@host:port),不包含请求参数 */
		uri normalize_uri()const
		{
			return simplemq::make_uri(host, port, password);
		}
	};
	/**
	 * redis 客户端(cpp_redis::client)的资源池管理器类型
	 */
	using redis_client_resource_manager_type = gdface::mt::resource_manager<std::shared_ptr<cpp_redis::client>>;
	/**
	 * redis 客户端(cpp_redis::client)的资源池管理器封装到std::shared_ptr
	 */
	using redis_client_resource_manager_ptr = std::shared_ptr<redis_client_resource_manager_type>;
	/**
	 * REDIS 消息订阅实现
	 */
	class RedisSubscriber : public virtual BaseSubscriber {
	private:
		cpp_redis::subscriber sub;
		mq_client_params params;
	protected:
		void doSubscribe(const std::vector<std::string>& channels) override
		{
			open();
			auto f = std::bind(&BaseSubscriber::dispatch, this, std::placeholders::_1, std::placeholders::_2);
			std::for_each(channels.begin(), channels.end(), [&](const std::string &c) {
				sub.subscribe(c, f);
			});
			sub.commit();
		}

		void doUnsubscribe(const std::vector<std::string>& channels) override
		{
			std::for_each(channels.begin(), channels.end(), [&](const std::string c) {
				sub.unsubscribe(c);
			});
			sub.commit();
		}

		void open() override
		{
			if (!sub.is_connected()) {
				sub.connect(params.host, params.port, [](const std::string& host, std::size_t port, cpp_redis::connect_state status) {
					switch (status) {
					case  cpp_redis::connect_state::dropped:
						SAMPLE_LOG("client disconnected from {}:{}", host, port);
						break;
					case cpp_redis::connect_state::start:
						SAMPLE_LOG("client connect to {}:{}", host, port);
						break;
					default:
						break;
					}
				},
					/** 超时(毫秒) */params.timeout_ms,
					/** 最大重连次数(-1为无限) */params.max_reconnects,
					/** 重连间隔(毫秒) */params.reconnect_interval_ms);

				if (!params.password.empty()) {
					//! authentication if server-server requires it
					sub.auth(params.password, [](const cpp_redis::reply& reply) {
						if (reply.is_error()) {
							SAMPLE_ERR("Authentication failed: {}", reply.as_string());
						}
						else {
							SAMPLE_LOG("successful authentication");
						}
					});
				}
			}
		}

	public:
		/**
		 * 构造函数
		 * @param params parameters for connect redis server
		 */
		RedisSubscriber(const mq_client_params &params)
			: params({params.host, params.port, params.password, DEFAULT_USERNAME,params.timeout_ms, DEFAULT_MAX_RECONNECTS, params.reconnect_interval_ms}) {}
		/**
		 * 构造函数
		 * @param uri redis uri
		 */
		RedisSubscriber(const uri &uri)
			: params(mq_client_params(uri)){}
		/**
		* 构造函数
		* @param params json parameters for connect mqtt server
		*/
		_BASIC_JSON_TPL_DECLARATION_
		RedisSubscriber(const _BASIC_JSON_TPL_ &params)
			: params(mq_client_params(params)){}
	public:
		RedisSubscriber() = default;
		/** 禁用移动构造函数 */
		RedisSubscriber(RedisSubscriber&&) = delete;
		/** 禁用拷贝构造函数 */
		RedisSubscriber(const RedisSubscriber&) = delete;
		/** 禁用赋值操作符 */
		RedisSubscriber& operator=(const RedisSubscriber&)  = delete;

		std::string to_string() const
		{
			std::stringstream ss;
			if (params.password.empty()) {
				ss << "redis://:" << params.host << ":" << params.port;
			}
			else {
				ss << "redis://:" << params.password << "@" << params.host << ":" << params.port;
			}
			return ss.str();
		}

	};

	/**
	 * cpp_redis::client 实例资源管理器
	 */
	struct redis_client_factory : gdface::mt::resource_factory<std::shared_ptr<cpp_redis::client>> {
		using do_init_type = std::function<void(cpp_redis::client&)>;
		/**
		 * @param do_init_fun client 初始化函数
		 */
		redis_client_factory(do_init_type do_init_fun) : do_init_fun(do_init_fun) {}
		~redis_client_factory() = default;
		virtual bool is_init(const resource_type&c) {
			return c && c->is_connected();
		}
		virtual void init(resource_type&c) override final {
			if (!c) {
				c = std::make_shared<cpp_redis::client>();
			}
			if (!is_init(c)) {
				do_init_fun(*c);
			}
		}
	private:
		do_init_type do_init_fun;
	};
	///////////////////////////////////////////
	/**
	 * 基于 redis 实现的消息发布接口
	 */
	class redis_publisher : public publisher {
	public:
		int64_t publish(const std::string &channel, const std::string& message) override
		{
			auto guard = rm->resource_guard();
			auto &client = (**guard);
			auto reply = client.publish(channel, message);
			client.sync_commit();
			return reply.get().as_integer();
		}
		redis_publisher(const redis_client_resource_manager_ptr &rm) : rm(rm) {
			assert(rm);
		}
		~redis_publisher() = default;
	private:
		redis_client_resource_manager_ptr rm;
	};
	/************************************************************************/
	/*            REDIS Client Manager 单实例管理                           */
	/*            必须以全局方式定义一个单实例                              */
	/************************************************************************/
	class ClientManager : public virtual BaseClientManager, public std::enable_shared_from_this<ClientManager> {
	public:
		ClientManager() = default;
		~ClientManager() = default;
		/* 禁用拷贝构造函数 */
		ClientManager(const ClientManager&) = delete;
		/* 禁用赋值操作符 */
		ClientManager& operator=(const ClientManager&) = delete;
		/**
		 * 创建redis 客户端资源管理器实例
		 * @param params 连接参数
		 * @param count count of instance
		 * @param connect_callback
		 */
		redis_client_resource_manager_ptr make_client_manager(const mq_client_params &params,
			size_t count = 1,
			const cpp_redis::connect_callback_t &connect_callback = nullptr)
		{
			std::lock_guard<std::mutex> guard(clts_lock);

			auto target = make_uri(params.host, params.port);
			if (!client_instances.count(target)) {
				// 不存在则创建新实例
				auto factory = std::make_shared<redis_client_factory>([=](cpp_redis::client&c) {
					c.connect(params.host, params.port, connect_callback, params.timeout_ms, params.max_reconnects, params.reconnect_interval_ms);
					if (!params.password.empty()) {
						c.auth(params.password);
					}
				});
				client_instances[target] = std::make_shared<redis_client_resource_manager_type>(count, factory);
			}
			return client_instances[target];
		}
		/**
		 * 创建 redis 客户端资源管理器实例
		 * @param uri uri for connect redis server,just like 'redis://127.0.0.1:6379'
		 * @param count count of instance
		 * @param connect_callback
		 */
		redis_client_resource_manager_ptr make_client_manager(const uri &uri,
			size_t count = 1,
			const cpp_redis::connect_callback_t &connect_callback = nullptr)
		{
			return make_client_manager(mq_client_params(uri), count, connect_callback);
		}

		/**
		 * 创建 redis 客户端资源管理器实例
		 * @param params parameters for connect redis server
		 * @param count count of instance
		 * @param connect_callback
		 */
		_BASIC_JSON_TPL_DECLARATION_
		redis_client_resource_manager_ptr make_client_manager(const _BASIC_JSON_TPL_ & params,
			size_t count = 1,
			const cpp_redis::connect_callback_t &connect_callback = nullptr)
		{
			return make_client_manager(mq_client_params(params), count, connect_callback);
		}
		/**
		 * 返回host,port指定的redis服务的RedisSubscriber实例引用,
		 * @param params parameters to be connected to
		 */
		subscriber_ptr getSubscriber(const mq_client_params& params) {
			std::lock_guard<std::mutex> guard(subs_lock);
			auto target = params.normalize_uri();
			if (!sub_instances.count(target)) {
				/** 不存在则创建新实例 */
				sub_instances[target] = std::make_shared<RedisSubscriber>(params);
			}
			return sub_instances[target];
		}
		/**
		 * 返回host,port指定的mqtt服务的MqttSubscriber实例引用,
		 * @param params parameters to be connected to
		 */
		subscriber_ptr getSubscriber(const base_client_params_ptr& params) override {
			auto ptr = std::dynamic_pointer_cast<mq_client_params>(params);
			throw_if_msg(!ptr, "params is not pointer of mq_client_params");
			return getSubscriber(*ptr);
		}
		/**
		 * 返回访问uri指定的redis服务的RedisSubscriber实例引用,
		 * 对于每一个不同的uri,返回对应的单实例
		 * @param uri uri for connect redis server,just like 'redis://127.0.0.1:6379'
		 */
		subscriber_ptr getSubscriber(const uri &uri) override
		{
			return getSubscriber(mq_client_params(uri));
		}
		/**
		 * 返回访问uri指定的redis服务的RedisSubscriber实例引用,
		 * 对于每一个不同的uri,返回对应的单实例
		 * @param params parameters for connect redis server
		 */
		subscriber_ptr getSubscriber(const simplemq::json &params) override
		{
			return getSubscriber(mq_client_params(params));
		}
		/**
		 * 创建 IPublisher 消息发布实例
		 * @param rm redis client 资源池实例
		 */
		publisher_ptr getPublisher(const std::shared_ptr<redis_client_resource_manager_type> &rm) {
			assert(rm);
			auto pub = std::make_shared<redis_publisher>(rm);
			return std::make_shared<IPublisher>(pub);
		}
		/**
		 * 创建 IPublisher 消息发布实例
		 * @param params parameters for connect redis server
		 */
		publisher_ptr getPublisher(const mq_client_params & params) {
			auto rm = make_client_manager(params);
			return getPublisher(rm);
		}
		/**
		 * 创建 IPublisher 消息发布实例
		 * @param params parameters for connect redis server
		 */
		publisher_ptr getPublisher(const base_client_params_ptr& params) override {
			auto ptr = std::dynamic_pointer_cast<mq_client_params>(params);
			throw_if_msg(!ptr, "params is not pointer of mq_client_params");
			return getPublisher(*ptr);
		}
		/**
		 * 创建 IPublisher 消息发布实例
		 * @param params parameters for connect mqtt server
		 */
		publisher_ptr getPublisher(const simplemq::json & params) override {
			auto rm = make_client_manager(params);
			return getPublisher(rm);
		}
		/**
		 * 创建 IPublisher 消息发布实例
		 * @param uri uri for connect mqtt server,just like 'redis://127.0.0.1:6379'
		 */
		publisher_ptr getPublisher(const uri&uri) override {
			auto rm = make_client_manager(uri);
			return getPublisher(rm);
		}
		std::string getImplType()const override
		{
			return "REDIS";
		}
		base_client_params_ptr asClientParam(const simplemq::json & params) override {
			return std::make_shared<mq_client_params>(params);
		}
	private:
		winsock_initializer winsock_init;
		std::map<uri, redis_client_resource_manager_ptr> client_instances;
		std::mutex clts_lock;
		std::map<uri, subscriber_ptr> sub_instances;
		std::mutex subs_lock;
	};
} /** namespace simplemq */
#endif /* SIMPLEMQ_REDIS_SIMPLEMQ_REDIS_HPP_ */
